package com.zongxingh.falcon.mq.consumer.impl;

import com.zongxingh.falcon.mq.config.MqProperties;
import com.zongxingh.falcon.mq.consumer.MqConsumer;
import com.zongxingh.falcon.mq.consumer.MsgConsume;
import com.zongxingh.falcon.mq.entity.SimpleMessage;
import com.zongxingh.falcon.mq.utils.ExceptionTools;
import com.zongxingh.falcon.mq.utils.MqUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * Introduction:  <简介>
 * Package Name:  com.zongxingh.falcon.mq.consumer.impl
 * Project Name:  mq
 * Author:  zongxingh
 * Email： zongxingh@163.com
 * Tel: 157*****778
 * Create Time:  2020/9/22 15:45
 */
public class RocketMqConsumer extends MqConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqConsumer.class);
    private DefaultMQPushConsumer defaultMQPushConsumer;


    public RocketMqConsumer() {
    }


    public RocketMqConsumer(String consumerGroup, String topic, MqProperties properties, MsgConsume msgConsume) {
        this.consumerGroup = consumerGroup;
        this.topic = topic;
        this.msgConsume = msgConsume;
        this.name = MqUtils.genMqConsumerName(consumerGroup,topic);
        this.defaultMQPushConsumer = createDefaultMQPushConsumer(properties);
    }

    public void setDefaultMQPushConsumer(DefaultMQPushConsumer defaultMQPushConsumer) {
        this.defaultMQPushConsumer = defaultMQPushConsumer;
    }

    public DefaultMQPushConsumer getDefaultMQPushConsumer() {
        return defaultMQPushConsumer;
    }


    private DefaultMQPushConsumer createDefaultMQPushConsumer(MqProperties properties) {
        // 判断当前 defaultMQPushConsumer 是否已经实例化，如果已经实例化则直接返回，否则新创建
        if (null != this.defaultMQPushConsumer) {
            LOGGER.info("创建RocketMq 消费者，该消费者已经存在，返回已创建的消费者");
            return this.defaultMQPushConsumer;
        }
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.consumerGroup);
        try {
            MqProperties.Rocketmq rocketmq = properties.getRocketmq();
            consumer.setNamesrvAddr(rocketmq.getNamesrvAddr());
            consumer.setClientCallbackExecutorThreads(rocketmq.getClientCallbackExecutorThreads());
            consumer.setConsumeThreadMax(rocketmq.getPushConsumer().getConsumeThreadMax());
            consumer.setConsumeThreadMin(rocketmq.getPushConsumer().getConsumeThreadMin());
            // 订阅Topics
            consumer.subscribe(this.topic, "*");
            consumer.setMessageModel(MessageModel.CLUSTERING);
            // 注册消息监听者
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {


                    List<SimpleMessage> messageList = new ArrayList<>();
                    for (MessageExt result : messages) {
                        String body = new String(result.getBody(), StandardCharsets.UTF_8);

                        SimpleMessage simpleMessage = new SimpleMessage();
                        simpleMessage.setTopic(result.getTopic());
                        simpleMessage.setTags(result.getTags());
                        simpleMessage.setMsgId(result.getMsgId());
                        simpleMessage.setMsg(body);
                        messageList.add(simpleMessage);
                    }
                    boolean isSuccess = msgConsume.consumeMessages(messageList);
                    if (isSuccess) return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            // 启动消费者
            consumer.start();
            LOGGER.info("消费创建完成，consumer group为：{}，topic为：{}", this.consumerGroup, this.topic);
        } catch (MQClientException e) {
            LOGGER.error("创建消费者发生异常，consumer group为：{}，topic为：{}，异常信为：\n{}", this.consumerGroup, this.topic,
                    ExceptionTools.getStackTraceAsString(e));
        }
        return consumer;
    }

}
